package com.hmall.transport.netty.service;

import com.hmall.enums.CompressTypeEnum;
import com.hmall.enums.RpcResponseCodeEnum;
import com.hmall.enums.SerializationTypeEnum;
import com.hmall.factory.SingletonFactory;
import com.hmall.remoting.constants.RpcContants;
import com.hmall.remoting.dto.RpcMessage;
import com.hmall.remoting.dto.RpcRequest;
import com.hmall.remoting.dto.RpcResponse;
import com.hmall.remoting.handler.RpcRequestHandler;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyRpcServerHandler extends ChannelInboundHandlerAdapter {

    private final RpcRequestHandler rpcRequestHandler;

    public NettyRpcServerHandler() {
        rpcRequestHandler = SingletonFactory.getInstance(RpcRequestHandler.class);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {//可读事件
        try {
            if (msg instanceof RpcMessage) {
                log.info("server receive msg: [{}] ", msg);
                byte messageType = ((RpcMessage) msg).getMessageType();
                RpcMessage rpcMessage = new RpcMessage();
                rpcMessage.setCodec(SerializationTypeEnum.HESSIAN.getCode());
                rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode());
                if (messageType == RpcContants.HEARTBEAT_REQUEST_TYPE){
                    rpcMessage.setMessageType(RpcContants.HEARTBEAT_RESPONSE_TYPE);
                    rpcMessage.setData(RpcContants.PONG);
                }else {
                    RpcRequest rpcRequest = (RpcRequest) ((RpcMessage) msg).getData();
                    Object result = rpcRequestHandler.handle(rpcRequest);
                    rpcMessage.setMessageType(RpcContants.RESPONSE_TYPE);
                    if (ctx.channel().isActive() && ctx.channel().isWritable()) {
                        RpcResponse<Object> rpcResponse = RpcResponse.success(result,rpcRequest.getRequestId());
                        rpcMessage.setData(rpcResponse);
                    }else {
                        RpcResponse<Object> fail = RpcResponse.fail(RpcResponseCodeEnum.FAIL);
                        rpcMessage.setData(fail);
                        log.error("not writable now, message dropped");
                    }
                }
                ctx.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);//当异步写入操作失败时，自动关闭 Channel
            }
        } finally {
            ReferenceCountUtil.release(msg);//手动释放ByteBuf，msg底层为ByteBuf直接内存
        }

    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {//用于心跳检测、空闲连接管理及业务逻辑触发
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                log.info("idle check happen, so close the connection");
                ctx.close();
            }
        }else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {//异常事件
        log.error("server catch exception");
        cause.printStackTrace();
        ctx.close();
    }
}
